/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.coders;



import com.bff.gaia.unified.sdk.values.PCollection;

import com.bff.gaia.unified.sdk.values.TypeDescriptor;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.ImmutableList;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



import javax.annotation.CheckForNull;

import java.lang.annotation.*;

import java.lang.reflect.InvocationTargetException;

import java.lang.reflect.Method;

import java.util.List;



/**

 * The {@link DefaultCoder} annotation specifies a {@link Coder} class to handle encoding and

 * decoding instances of the annotated class.

 *

 * <p>The specified {@link Coder} must have the following method:

 *

 * <pre>

 * {@code public static CoderProvider getCoderProvider()}.

 * </pre>

 *

 * <p>Coders specified explicitly via {@link PCollection#setCoder} take precedence, followed by

 * Coders found at runtime via {@link CoderRegistry#getCoder}. See {@link CoderRegistry} for a more

 * detailed discussion of the precedence rules.

 */

@Documented

@Retention(RetentionPolicy.RUNTIME)

@Target(ElementType.TYPE)

@SuppressWarnings("rawtypes")

public @interface DefaultCoder {

  @CheckForNull

  Class<? extends Coder> value();



  /**

   * A {@link CoderProviderRegistrar} that registers a {@link CoderProvider} which can use the

   * {@code @DefaultCoder} annotation to provide {@link CoderProvider coder providers} that creates

   * {@link Coder}s.

   */

  class DefaultCoderProviderRegistrar implements CoderProviderRegistrar {



    @Override

    public List<CoderProvider> getCoderProviders() {

      return ImmutableList.of(new DefaultCoderProvider());

    }



    /**

     * A {@link CoderProvider} that uses the {@code @DefaultCoder} annotation to provide {@link

     * CoderProvider coder providers} that create {@link Coder}s.

     */

    static class DefaultCoderProvider extends CoderProvider {

      private static final Logger LOG = LoggerFactory.getLogger(DefaultCoderProvider.class);



      /**

       * Returns the {@link Coder} returned according to the {@link CoderProvider} from any {@link

       * DefaultCoder} annotation on the given class.

       */

      @Override

      public <T> Coder<T> coderFor(

		  TypeDescriptor<T> typeDescriptor, List<? extends Coder<?>> componentCoders)

          throws CannotProvideCoderException {



        Class<?> clazz = typeDescriptor.getRawType();

        DefaultCoder defaultAnnotation = clazz.getAnnotation(DefaultCoder.class);

        if (defaultAnnotation == null) {

          throw new CannotProvideCoderException(

              String.format("Class %s does not have a @DefaultCoder annotation.", clazz.getName()));

        }



        Class<? extends Coder> defaultAnnotationValue = defaultAnnotation.value();

        if (defaultAnnotationValue == null) {

          throw new CannotProvideCoderException(

              String.format(

                  "Class %s has a @DefaultCoder annotation with a null value.", clazz.getName()));

        }



        LOG.debug(

            "DefaultCoder annotation found for {} with value {}", clazz, defaultAnnotationValue);



        Method coderProviderMethod;

        try {

          coderProviderMethod = defaultAnnotationValue.getMethod("getCoderProvider");

        } catch (NoSuchMethodException e) {

          throw new CannotProvideCoderException(

              String.format(

                  "Unable to find 'public static CoderProvider getCoderProvider()' on %s",

                  defaultAnnotationValue),

              e);

        }



        CoderProvider coderProvider;

        try {

          coderProvider = (CoderProvider) coderProviderMethod.invoke(null);

        } catch (IllegalAccessException

            | IllegalArgumentException

            | InvocationTargetException

            | NullPointerException

            | ExceptionInInitializerError e) {

          throw new CannotProvideCoderException(

              String.format(

                  "Unable to invoke 'public static CoderProvider getCoderProvider()' on %s",

                  defaultAnnotationValue),

              e);

        }

        return coderProvider.coderFor(typeDescriptor, componentCoders);

      }

    }

  }

}